
/***************************************************************************
 *   sakcab 
 *                                                                         *
 *   Copyright (C) 2010 by kuerant                                         *
 *                         <kuerant@gmail.com>                             *
 *                                                                         *
 *   This library is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU Library General Public License as       *
 *   published by the Free Software Foundation; either version 2 of the    *
 *   License, or (at your option) any later version.                       *
 *                                                                         *
 *   This library is distributed in the hope that it will be usefull, but  *
 *   WITHOUT ANY WARRANTY; without even the implied warranty of            *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU      *
 *   Library General Public License for more details.                      *
 *                                                                         *
 *   You should have received a copy of the GNU Library General Public     *
 *   License along with this library; if not, write to the Free Software   *
 *   Foundation, Inc., 59 Temple Place - Suite 330, Boston,                *
 *   MA 02111-1307, USA                                                    *
 ***************************************************************************/


#include <unistd.h>
#include <iostream>
#include <fstream>
#include <string>
#include <vector>

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>

#include <boost/program_options.hpp>
namespace po = boost::program_options;

#include <boost/date_time/posix_time/posix_time.hpp>
using namespace boost::posix_time;

#include <pthread.h>
#include <signal.h>

#include "bocat/common/base64.hpp"
#include "bocat/common/logging.hpp"
#include "bocat/common/time.hpp"
#include "bocat/common/url.hpp"

#include "sakcab/client.hpp"


static bool	sg_will_stop = false;

static sakcab::client::Client*	sg_client_p = NULL;

bocat::common::Logger*    logger_p = NULL;

#define INFO(ARGS...) 	logger_p->info(__LINE__, __FILE__, ARGS)
#define ERROR(ARGS...) 	logger_p->error(__LINE__, __FILE__, ARGS)


void	handleStopSignal( int signal )
{
	sg_will_stop = true;
}


float
getTimeDuration( ptime time_begin, ptime time_end )
{
    time_period tp( time_begin, time_end );
    time_duration   td = tp.length();
    long    time_last_ms = td.total_milliseconds();
    float   time_last_s = time_last_ms / 1000.0f;

    return time_last_s;
}


const size_t	MAX_BATCH_SIZE = 1024 * 1024 * 10;	// 10 MB

// load file in batch mode
int	loadFile( const std::string& filename )
{
	boost::filesystem::path	file_path( filename );
	if ( !boost::filesystem::exists( file_path ) ) {
		// file does NOT exists, just ignore it
		return	0;
	}

	int		rc = 0;

	INFO( " - loading %s ...", filename.c_str() );
	//std::cout << bocat::common::now() << " - loading " << filename << " ...\n";
	unsigned	count_bad_records = 0;
	unsigned	count_records = 0;
	unsigned	count_bytes = 0;

	std::vector< sakcab::client::kv_type >	kvs;
	size_t		sum_kvs = 0;

	ptime   time_start = microsec_clock::local_time();
	std::ifstream	ifs( filename.c_str() );
	while ( !ifs.eof() ) {
		if (sg_will_stop) {
			break;
		}

		std::string	line;
		std::getline( ifs, line );
		boost::trim( line );
		if ( line.size() < 1 ) {
			//empty line
			continue;
		}

		std::vector< std::string >	fields;
		boost::split(fields, line, boost::is_any_of("\t"));
		if ( fields.size() < 3 ) {
			// bad format
			ERROR( "BAD format data : %s", line.c_str() );
			//std::cerr << "BAD format : " << line << std::endl;
			++count_bad_records;
			continue;
		}

		// fields : timestamp \t url \t data
		const std::string&	url_s = fields[ 1 ];
		if ( url_s.size() < 3 ) {
			// bad URL 
			ERROR( "BAD URL data : %s", line.c_str() );
			//std::cerr << "BAD URL : " << line << std::endl;
			++count_bad_records;
			continue;
		}
		
		std::string	url_reversed;
		try  {
			bocat::common::URL	url( url_s );
			url_reversed = url.getReversedURL();
			boost::trim( url_reversed );
		}
		catch (std::exception& e) {
			ERROR( "Exception : %s", e.what() );
			ERROR( "BAD LINE : %s", line.c_str() );
			//std::cerr << "Exception : " << e.what() << std::endl;
			//std::cerr << "BAD LINE : " << line << std::endl;
			++count_bad_records;
			continue;
		}

		if ( fields.size() > 3 ) {
			std::string	data( fields[ 2 ] );
			for (unsigned k=3; k<fields.size(); ++k) {
				data.append( " " );
				data.append( fields[ k ] );
			}

			sakcab::client::kv_type	kv( url_reversed, data );
			kvs.push_back( kv );
			sum_kvs += data.size();

			count_bytes += data.size();
		}
		else {
			const std::string&	data = fields[ 2 ];

			sakcab::client::kv_type	kv( url_reversed, data );
			kvs.push_back( kv );
			sum_kvs += data.size();

			count_bytes += data.size();
		}

		++count_records;

		if ( sum_kvs < MAX_BATCH_SIZE ) {
			continue;
		}

		try {
			rc = sg_client_p->insertAppend( kvs );
			kvs.clear();
			sum_kvs = 0;
		}
		catch( std::exception& e ) {
			ERROR( " * load %s failed -- %s", filename.c_str(), e.what() );
			//std::cerr << bocat::common::now() << " * load " << filename << " failed -- " << e.what() << std::endl;
			rc = -1;
		}
		if ( rc < 0 ) {
			ifs.close();
			INFO( " * load %s failed", filename.c_str() );
			//std::cout << bocat::common::now() << " * load " << filename << " failed.\n";
			return	-1;
		}
	}
	ifs.close();

	if ( sum_kvs > 0 ) {
		// send the rest data
		try {
			rc = sg_client_p->insertAppend( kvs );
			kvs.clear();
			sum_kvs = 0;
		}
		catch( std::exception& e ) {
			ERROR( " * load %s failed -- %s", filename.c_str(), e.what() );
			//std::cerr << bocat::common::now() << " * load " << filename << " failed -- " << e.what() << std::endl;
			rc = -1;
		}
		if ( rc < 0 ) {
			ifs.close();
			INFO( " * load %s failed", filename.c_str() );
			//std::cout << bocat::common::now() << " * load " << filename << " failed.\n";
			return	-1;
		}
	}

	ptime   time_stop = microsec_clock::local_time();
	float   time_last_s = getTimeDuration( time_start, time_stop );
	float   tps = count_records / time_last_s;
	float   bw = count_bytes / time_last_s / 1048576.0f;
	INFO( " * load %s done (%u records last %.2f seconds, tps=%.2f, bw=%.2f MB/s, bad-records:%u", filename.c_str(), count_records, time_last_s, tps, bw, count_bad_records );
	//std::cout << bocat::common::now() << " * load " << filename << " done (" << count_records << " records last "  << std::setprecision(3) << std::fixed << time_last_s << " seconds, tps=" << tps << ", bw=" << bw << " MB/s" << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << ", bad-records:" << count_bad_records << ").\n";
	return	0;
}


int	watchFileList( const std::string& filelist )
{
	std::string	filelist_running = filelist + "-running";
	boost::filesystem::path	running_path( filelist_running );
	boost::filesystem::path	filelist_path( filelist );

	while ( !sg_will_stop ) {
		// handle running filelist-path
		if ( boost::filesystem::exists(running_path) ) {
			// handle running filelist
			std::ifstream	ifs(filelist_running.c_str());
			while ( !ifs.eof() ) {
				if (sg_will_stop) {
					break;
				}

				std::string	filename;
				ifs >> filename;
				boost::trim( filename );
				if ( filename.size() < 1 ) {
					continue;
				}

				int	rc = loadFile( filename );
				if ( rc < 0 ) {
					ifs.close();
					//std::cerr << bocat::common::now() << " fail to load file " << filename << std::endl;
					ERROR( " * fail to load file %s", filename.c_str() );
					return	-1;
				}
				if (sg_will_stop) {
					// may be load fail  because of signal to sleep
					break;
				}

				// rename file after loading
				std::string	filename_load = filename + ".bak";
				rc = rename( filename.c_str(), filename_load.c_str() );
				if ( rc < 0 ) {
					ERROR( " * fail to rename file %s to %s", filename.c_str(), filename_load.c_str() );
				}

				// delete file after loading
				//unlink( filename.c_str() );
			}
			ifs.close();

			// all done, just deleting running_path
			boost::filesystem::remove( running_path );
		}

		// handle filelist
		if ( boost::filesystem::exists(filelist_path) ) {
			// rename to running filelist
			boost::filesystem::rename( filelist_path, running_path );
		}
		else {
			// no filelist, just sleep
			sleep( 60 );
		}
	}

	return	0;
}


int main(int argc, char* argv[])
{
    std::string host;
    std::string port;

	std::string	filelist;
	std::string	file;

	std::string	log_config;
	std::string	log_section;


    try {
        po::options_description desc("Allowed options");
        desc.add_options()
            ("help,h", "produce help message")
            ("host", po::value<std::string>(&host)->default_value("127.0.0.1"), "server ip/hostname")
            ("port", po::value<std::string>(&port)->default_value("5137"), "server port")

            ("file", po::value<std::string>(&file), "data file name")
            ("filelist", po::value<std::string>(&filelist), "path to data filelist")

			("log-config", po::value<std::string>(&log_config), "path to logging config file")
			("log-section", po::value<std::string>(&log_section), "section name of logging config")
        ;

        po::variables_map vm;
        po::store(po::parse_command_line(argc, argv, desc), vm);
        po::notify(vm);

        if (vm.count("help")) {
            std::cout << desc << "\n";
            return 1;
        }
		
		if ((vm.count("file") + vm.count("filelist")) < 1) {
			std::cout << "\none of data file or filelist must be provided\n\n";
            std::cout << desc << "\n";
            return 1;
		}

		if ((vm.count("log-config") > 0) && (vm.count("log-section") < 1)) {
			std::cout << "\nlog-section must be provided\n\n";
            std::cout << desc << "\n";
            return 2;
		}
    }
    catch(std::exception& e) {
        std::cerr << "error: " << e.what() << "\n";
        return 1;
    }
    catch(...) {
        std::cerr << "Exception of unknown type!\n";
        return 2;
    }

	// handle signal 
	struct sigaction        sa;

	sigemptyset(&sa.sa_mask);
	sa.sa_flags = 0;
	sa.sa_handler = handleStopSignal;

	sigaction(SIGHUP, &sa, NULL);
	sigaction(SIGINT, &sa, NULL);
	sigaction(SIGQUIT, &sa, NULL);
	sigaction(SIGPIPE, &sa, NULL);
	sigaction(SIGTERM, &sa, NULL);


	logger_p = bocat::common::getLogger();
	if ( log_config.size() > 0 ) {
		logger_p->configure( log_config.c_str(), log_section.c_str() );
	}


	// prepare client
	//std::cout << bocat::common::now() << " connect to " << host << ":" << port << " ...\n";
	int	port_i = boost::lexical_cast< int >( port.c_str() );
	INFO( " connect to %s:%d ...", host.c_str(), port_i );
	sg_client_p	= new sakcab::client::Client( host.c_str(), port_i );

	int	rc = 0;
	if (file.size() > 0) {
		rc = loadFile( file );
	}
	else {
		rc = watchFileList( filelist );
	}

	
	//std::cout << bocat::common::now() << " shutdown ...\n";
	INFO( " shutdown ..." );

	delete sg_client_p;
	sg_client_p = NULL;

	//std::cout << bocat::common::now() << " done.\n";
	INFO( " done." );

	return 0;
}


///////////////////////////////////////////////////////////////////////

/*
// load file in line-by-line mode
int	loadFile( const std::string& filename )
{
	boost::filesystem::path	file_path( filename );
	if ( !boost::filesystem::exists( file_path ) ) {
		// file does NOT exists, just ignore it
		return	0;
	}

	int		rc = 0;

	INFO( " - loading %s ...", filename.c_str() );
	//std::cout << bocat::common::now() << " - loading " << filename << " ...\n";
	unsigned	count_bad_records = 0;
	unsigned	count_records = 0;
	unsigned	count_bytes = 0;
	ptime   time_start = microsec_clock::local_time();
	std::ifstream	ifs( filename.c_str() );
	while ( !ifs.eof() ) {
		if (sg_will_stop) {
			break;
		}

		std::string	line;
		std::getline( ifs, line );
		boost::trim( line );
		if ( line.size() < 1 ) {
			//empty line
			continue;
		}

		std::vector< std::string >	fields;
		boost::split(fields, line, boost::is_any_of("\t"));
		if ( fields.size() < 3 ) {
			// bad format
			ERROR( "BAD format data : %s", line.c_str() );
			//std::cerr << "BAD format : " << line << std::endl;
			++count_bad_records;
			continue;
		}

		// fields : timestamp \t url \t data
		std::string	url_reversed;
		const std::string&	url_s = fields[ 1 ];
		if ( url_s.size() < 3 ) {
			// bad URL 
			ERROR( "BAD URL data : %s", line.c_str() );
			//std::cerr << "BAD URL : " << line << std::endl;
			++count_bad_records;
			continue;
		}
		
		try  {
			bocat::common::URL	url( url_s );
			url_reversed = url.getReversedURL();
		}
		catch (std::exception& e) {
			ERROR( "Exception : %s", e.what() );
			ERROR( "BAD LINE : %s", line.c_str() );
			//std::cerr << "Exception : " << e.what() << std::endl;
			//std::cerr << "BAD LINE : " << line << std::endl;
			++count_bad_records;
			continue;
		}

		if ( fields.size() > 3 ) {
			std::string	data( fields[ 2 ] );
			for (unsigned k=3; k<fields.size(); ++k) {
				data.append( " " );
				data.append( fields[ k ] );
			}
			count_bytes += data.size();
			try {
				rc = sg_client_p->insertAppend( url_reversed, data );
			}
			catch( std::exception& e ) {
				ERROR( " * load %s failed -- %s", filename.c_str(), e.what() );
				//std::cerr << bocat::common::now() << " * load " << filename << " failed -- " << e.what() << std::endl;
				rc = -1;
			}
		}
		else {
			const std::string&	data = fields[ 2 ];
			count_bytes += data.size();
			try {
				rc = sg_client_p->insertAppend( url_reversed, data );
			}
			catch( std::exception& e ) {
				ERROR( " * load %s failed -- %s", filename.c_str(), e.what() );
				//std::cerr << bocat::common::now() << " * load " << filename << " failed -- " << e.what() << std::endl;
				rc = -1;
			}
		}
		if ( rc < 0 ) {
			ifs.close();
			INFO( " * load %s failed", filename.c_str() );
			//std::cout << bocat::common::now() << " * load " << filename << " failed.\n";
			return	-1;
		}

		++count_records;
	}
	ifs.close();

	ptime   time_stop = microsec_clock::local_time();
	float   time_last_s = getTimeDuration( time_start, time_stop );
	float   tps = count_records / time_last_s;
	float   bw = count_bytes / time_last_s / 1048576.0f;
	INFO( " * load %s done (%u records last %.2f seconds, tps=%.2f, bw=%.2f MB/s, bad-records:%u", filename.c_str(), count_records, time_last_s, tps, bw, count_bad_records );
	//std::cout << bocat::common::now() << " * load " << filename << " done (" << count_records << " records last "  << std::setprecision(3) << std::fixed << time_last_s << " seconds, tps=" << tps << ", bw=" << bw << " MB/s" << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << ", bad-records:" << count_bad_records << ").\n";
	return	0;
}

*/

